#include "store.h"

namespace crawler { namespace server {
Store::Store(const std::string &store_db_path) : store_db_path_(store_db_path) {
  // initialize level db
  leveldb::Options options;
  options.create_if_missing = true;
	options.write_buffer_size = 256 * 1048576;
  leveldb::DB *db;
  leveldb::Status status = leveldb::DB::Open(options, store_db_path_, &db);
  if (!status.ok()) {
    throw std::runtime_error("Failed to open leveldb, path=" + store_db_path_
                              + " error=" + status.ToString());
  }
  db_.reset(db);

  // initialize bloom filter
  bloom_parameters parameters;
  parameters.projected_element_count    = 2^21;
  parameters.false_positive_probability = (double)1/(1<<21);
  parameters.random_seed                = time(NULL);
  if (!parameters) {
    throw std::runtime_error("Invalid set of bloom filter parameters");
  }
  parameters.compute_optimal_parameters();
  found_links_filter_.reset(new bloom_filter(parameters));

  // insert existing values to bloom filter
  for (PageIterator it=Begin(); it!=End(); ++it) {
    found_links_filter_->insert(it->url);
  }
}
void Store::StorePage(const Page &page) {
  std::vector<Page> pages;
  if (page.to_url.empty()) {
    pages.push_back(page);
  } else {
    Page p1, p2(page);
    p1.__set_url(page.url);
    p1.__set_status(PageStatus::kRedirected);
    p2.__set_url(page.to_url);
    p2.__set_to_url(std::string());

    pages.push_back(p1);
    pages.push_back(p2);
  };
  for (std::vector<Page>::iterator it=pages.begin(); it!=pages.end(); ++it) {
    shared_ptr<TMemoryBuffer> mem_buf(new TMemoryBuffer);
    shared_ptr<TBinaryProtocol> bin_proto(new TBinaryProtocol(mem_buf));

    it->write(bin_proto.get());
    uint8_t* buf_ptr;
    uint32_t sz;
    mem_buf->getBuffer(&buf_ptr, &sz);

    leveldb::Slice value(reinterpret_cast<const char*>(buf_ptr), sz);
    leveldb::Status s = db_->Put(leveldb::WriteOptions(), it->url, value);
    if (!s.ok()) {
      throw std::runtime_error("Failed to insert into leveldb, key=" + it->url +
                               "error=" + s.ToString());
    }
  }
}
void Store::StoreLink(const Link &link) {
  if (!found_links_filter_->contains(link.url)) {
    std::cout << "store link, " << link.url << std::endl;
    found_links_filter_->insert(link.url);

    Page page;
    page.__set_url(link.url);
    page.__set_status(PageStatus::kNotYetDownload);
    StorePage(page);
  }
}
void Store::GetPage(const std::string &url, Page &page) {
  ServerError error;
  shared_ptr<TMemoryBuffer> mem_buf(new TMemoryBuffer);
  shared_ptr<TBinaryProtocol> bin_proto(new TBinaryProtocol(mem_buf));

  std::string value;
  leveldb::Status s = db_->Get(leveldb::ReadOptions(), url, &value);
  if (!s.ok()) {
    error.__set_why("Read leveldb error, key=" + url + 
                    " error=" + s.ToString());
    throw error;
  }
  mem_buf->resetBuffer((uint8_t*)value.data(), value.size());
  page.read(bin_proto.get());
}
Store::PageIterator Store::Begin() const {
  leveldb::Iterator* iter = db_->NewIterator(leveldb::ReadOptions());
  return PageIterator(boost::shared_ptr<leveldb::Iterator>(iter));
}
const Store::PageIterator &Store::End() const {
  return end_iter_;
}
}}

